package com.zhangtao.demo.config.redis.nettyclient.handler;

import java.util.ArrayList;
import java.util.List;

import com.zhangtao.demo.common.dto.Holder;
import com.zhangtao.demo.config.redis.nettyclient.NettyClient;
import com.zhangtao.demo.config.redis.nettyclient.command.RedisCommand;

import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.CodecException;
import io.netty.handler.codec.redis.ArrayRedisMessage;
import io.netty.handler.codec.redis.ErrorRedisMessage;
import io.netty.handler.codec.redis.FullBulkStringRedisMessage;
import io.netty.handler.codec.redis.IntegerRedisMessage;
import io.netty.handler.codec.redis.RedisMessage;
import io.netty.handler.codec.redis.SimpleStringRedisMessage;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;

public class RedisClientHandler extends ChannelDuplexHandler {

	// 发送 redis 命令
	@Override
	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
		String[] commands = ((String) msg).split("\\s+");
		List<RedisMessage> children = new ArrayList<>(commands.length);
		for (String cmdString : commands) {
			children.add(new FullBulkStringRedisMessage(ByteBufUtil.writeUtf8(ctx.alloc(), cmdString)));
		}
		RedisMessage request = new ArrayRedisMessage(children);
		ctx.write(request, promise);
	}

	// socket连接建立成功
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		ChannelFuture authCallback = null;
		if (ctx.channel().hasAttr(NettyClient.redisAuth)) {
			authCallback = ctx.channel().writeAndFlush(RedisCommand.auth(ctx.channel().attr(NettyClient.redisAuth).get()));
		}
		int db = ctx.channel().attr(NettyClient.redisdb).get();
		if (db < 0) {
			db = 0;
		}
		if (authCallback != null) {
			Holder<Integer> dbHolder = new Holder<>(db);
			authCallback.addListener(gfl -> {
				if (gfl.isDone()) {
					ctx.channel().writeAndFlush(RedisCommand.select(dbHolder.value()));
				}
			});
		} else {
			ctx.channel().writeAndFlush(RedisCommand.select(db));
		}
		super.channelActive(ctx);
	}

	// 接收 redis 响应数据
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
		RedisMessage redisMessage = (RedisMessage) msg;
		// 打印响应消息
		printAggregatedRedisResponse(redisMessage);
		// 是否资源
		ReferenceCountUtil.release(redisMessage);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		System.err.println("exceptionCaught: ");
		cause.printStackTrace(System.err);
		ctx.close();
	}

	private static void printAggregatedRedisResponse(RedisMessage msg) {
		if (msg instanceof SimpleStringRedisMessage) {
			System.out.println("SimpleStringRedisMessage   "+((SimpleStringRedisMessage) msg).content());
		} else if (msg instanceof ErrorRedisMessage) {
			System.out.println("ErrorRedisMessage   "+((ErrorRedisMessage) msg).content());
		} else if (msg instanceof IntegerRedisMessage) {
			System.out.println("IntegerRedisMessage   "+((IntegerRedisMessage) msg).value());
		} else if (msg instanceof FullBulkStringRedisMessage) {
			System.out.println("FullBulkStringRedisMessage   "+getString((FullBulkStringRedisMessage) msg));
		} else if (msg instanceof ArrayRedisMessage) {
			for (RedisMessage child : ((ArrayRedisMessage) msg).children()) {
				printAggregatedRedisResponse(child);
			}
		} else {
			throw new CodecException("unknown message type: " + msg);
		}
	}

	private static String getString(FullBulkStringRedisMessage msg) {
		if (msg.isNull()) {
			return "(null)";
		}
		return msg.content().toString(CharsetUtil.UTF_8);
	}
}
